package com.abyss.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * 从kafka中读取数据
 *
 * Kafka相关操作
 * 创建topic
 * ./bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic kafkatopic --zookeeper node1:2181
 * 模拟生产者
 * ./bin/kafka-console-producer.sh --broker-list node1:9092 --topic kafkatopic
 * 模拟消费者
 * ./bin/kafka-console-consumer.sh --from-beginning --topic kafkatopic --zookeeper node1:2181
 */
public class KafkaSourceDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Source
        Properties properties = new Properties();
        properties.setProperty("group.id", "abyss");
        properties.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        properties.setProperty("auto.offset.reset", "earliest");
        String topic = "kafkatopic";

        // 这一步得到的不是DataStreamSource
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                topic,
                new SimpleStringSchema(),
                properties
        );

        /*
            kafka的source对象可以指定从哪里开始消费
            如:
            kafkaConsumer.setStartFromEarliest();       // 从头开始消费
            kafkaConsumer.setStartFromTimestamp(System.currentTimeMillis()); // 从指定的时间戳开始消费
            kafkaConsumer.setStartFromGroupOffsets();   // 从group 中记录的offset开始消费
            kafkaConsumer.setStartFromLatest();         // 从最新开始消费

            以及指定每个从某个topic的某个分区的某个offset开始消费
            Map<KafkaTopicPartition, Long> offsets = new HashMap<>();
            offsets.put(new KafkaTopicPartition(topic, 0), 0L);
            offsets.put(new KafkaTopicPartition(topic, 1), 0L);
            offsets.put(new KafkaTopicPartition(topic, 2), 0L);
            kafkaConsumer.setStartFromSpecificOffsets(offsets);
            如上, 就指定了topic的分区0,1,2 都分别从offset 0 开始消费.
         */

        DataStreamSource<String> kafkaSource = env.addSource(kafkaConsumer).setParallelism(6);

        // 打印
        kafkaSource.print();

        env.execute();
    }
}